package org.sample.mqtt.service.impl;

import org.sample.mqtt.component.annotations.Topic;
import org.sample.mqtt.component.model.MqttRequest;
import org.sample.mqtt.component.model.MqttResponse;
import org.sample.mqtt.endpoint.MqttProducer;
import org.sample.mqtt.service.MessageService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * Created by Alan on 2019/8/7.
 */
@Service
public class MessageServiceImpl implements MessageService {

    private Map<String, CompletableFuture<MqttResponse>> topicSubscribers = new ConcurrentHashMap<>();

    @Autowired
    private MqttProducer mqttProducer;

    @Override
    public void notify(String deviceId, Object payload) {
        Topic annotation = payload.getClass().getAnnotation(Topic.class);
        String topic = annotation.value().replace("{deviceId}", deviceId);
        mqttProducer.sendTo(topic, payload);
    }

    @Override
    public CompletableFuture<MqttResponse> request(String deviceId, MqttRequest payload) {
        long serialNo = payload.getSerialNo();
        if (serialNo == 0L)
            payload.setSerialNo(serialNo = System.currentTimeMillis());

        CompletableFuture<MqttResponse> receive = subscribe(deviceId, serialNo);
        notify(deviceId, payload);
        return receive;
    }

    @Override
    public boolean response(Message<MqttResponse> message) {
        MqttResponse payload = message.getPayload();
        CompletableFuture<MqttResponse> future = topicSubscribers.remove(getKey(payload.getDeviceId(), payload.getResponseSerialNo()));
        if (future != null) {
            future.complete(payload);
            return true;
        }
        return false;
    }

    private String getKey(String deviceId, long serialNo) {
        return deviceId + "/" + serialNo;
    }

    private CompletableFuture<MqttResponse> subscribe(String deviceId, long serialNo) {
        String key = getKey(deviceId, serialNo);
        return topicSubscribers.computeIfAbsent(key, k -> new CompletableFuture<MqttResponse>()
                .orTimeout(5, TimeUnit.SECONDS)
                .whenComplete((mqttResponse, throwable) -> topicSubscribers.remove(k)));
    }
}